package com.personal.gateway.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.personal.gateway.constant.FlinkCommandConstantOld;
import com.personal.gateway.config.OssConfig;
import com.personal.gateway.config.ValueConfig;
import com.personal.gateway.dao.SavePointDao;
import com.personal.gateway.entity.FlinkCommandHistory;
import com.personal.gateway.entity.FlinkGatewayRoute;
import com.personal.gateway.mapper.FlinkCommandHistoryMapper;
import com.personal.gateway.mapper.FlinkGatewayRouteMapper;
import com.personal.gateway.service.JobService;
import com.personal.gateway.utils.StringTemplateUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
@Service
public class JobServiceImpl implements JobService {

    @Resource
    private ValueConfig valueConfig;

    @Resource
    private FlinkGatewayRouteMapper flinkGatewayRouteMapper;

    @Resource
    private FlinkCommandHistoryMapper flinkCommandHistoryMapper;

    @Resource
    private OssConfig ossConfig;

    @Override
    public String startNoSavePoint(String name) {
        FlinkGatewayRoute route = flinkGatewayRouteMapper.getRouteByRouteId(name);
        Map<String, String> parameter = initStartParameter(route);
        return StringTemplateUtils.render(FlinkCommandConstantOld.START_NO_SAVE_POINT, parameter);
    }

    @Override
    public String startWithSavePoint(String jobName, String filePath) {
        FlinkGatewayRoute route = flinkGatewayRouteMapper.getRouteByRouteId(jobName);
        Map<String, String> parameter = initStartParameter(route, filePath);
        return StringTemplateUtils.render(FlinkCommandConstantOld.START_WITH_SAVE_POINT, parameter);
    }

    @Override
    public String stopWithSavePoint(String jobName) {
        FlinkGatewayRoute route = flinkGatewayRouteMapper.getRouteByRouteId(jobName);
        Map<String, String> parameter = initStartParameter(route);
        return StringTemplateUtils.render(FlinkCommandConstantOld.STOP_WITH_SAVE_POINT, parameter);
    }

    /**
     * 初始启动参数数据
     */
    private Map<String, String> initStartParameter(FlinkGatewayRoute route) {
       return initStartParameter(route, "");
    }

    private Map<String, String> initStartParameter(FlinkGatewayRoute route, String filePath) {
        Map<String, String> parameter = new HashMap<>();
        // Flink版本
        parameter.put("flink_version", route.getFlinkVersion().substring(0, route.getFlinkVersion().indexOf("-")));
        // 启动类
        parameter.put("start_class", route.getStartClass());
        // 集群名称
        parameter.put("cluster_id", route.getRouteId());
        // 镜像名称
        parameter.put("docker_image", route.getImageName() + ":k8s-" + valueConfig.getActiveProfile());
        // taskManager内存
        parameter.put("task_manager_memory", route.getTaskManagerMemory());
        // K8s高可用备份点
        parameter.put("high_env", valueConfig.getActiveProfile());
        parameter.put("high_name", route.getRouteId());
        // 启动jar包
        parameter.put("start_jar", route.getStartJarName());
        // 启动环境变量
        parameter.put("start_env", valueConfig.getActiveProfile());
        // 启动 save point 文件夹名称
        parameter.put("start_save_point_path", filePath);
        // 停止save point 位置
        parameter.put("stop_save_point_path", FlinkCommandConstantOld.SAVE_POINT_PREFIX);
        parameter.put("stop_save_point_env", valueConfig.getActiveProfile());
        parameter.put("stop_save_point_name", route.getRouteId());
        return parameter;
    }

    @Override
    public String execute(JSONObject obj, String user) {
        Integer id = obj.getInteger("id");
        Integer type = obj.getInteger("type");
        String[] command = obj.getString("command").split("\n");
        String output = runExec(command);

        // 插入操作记录
        String commandString = StringUtils.join(command, " ");
        FlinkGatewayRoute route = flinkGatewayRouteMapper.getRouteById(id);
        FlinkCommandHistory history = new FlinkCommandHistory(route.getGatewayNum(), commandString, output, type, user);
        flinkCommandHistoryMapper.insert(history);
        return output;
    }

    @Override
    public List<SavePointDao> getSavePoint(String jobName) {
        return ossConfig.getSavePointList(valueConfig.getActiveProfile(), jobName);
    }

    private String runExec(String[] command) {
        try {
            String commandString = StringUtils.join(command, " ");
            Runtime runtime = Runtime.getRuntime();
            Process process = runtime.exec(command);

            try (BufferedReader outPutBufferedReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
                 BufferedReader errorPutBufferedReader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {

                int status = process.waitFor();

                StringBuilder outPutStringBuilder = new StringBuilder();
                String line;

                if (status != 0) {
                    while ((line = errorPutBufferedReader.readLine()) != null) {
                        outPutStringBuilder.append(line).append("\n");
                    }
                    return "执行参数: \n" + commandString + "\n执行异常,异常信息: \n" + outPutStringBuilder.toString();
                }
                while ((line = outPutBufferedReader.readLine()) != null) {
                    outPutStringBuilder.append(line).append("\n");
                }
                return "执行参数: \n" + commandString + "\n运行成功： \n" + outPutStringBuilder.toString();
            } catch (Exception e) {
                log.error("", e);
                return e.getMessage();
            } finally {
                if (process != null) {
                    process.destroy();
                }
            }
        } catch (IOException e) {
            log.error("", e);
            return e.getMessage();
        }
    }

}
